package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.entity.{COMPLETED_PAYMENT, PENDING_PAYMENT, TOPIC}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._


/**
  * 坑位产品订单
  *
  */
object WirelessPitOrder {

  import Conversions._

  def process(sc: SparkContext, pageForward: DataFrame): Unit = {

    val commons = Seq("sessionId", "deviceToken", "channelCode", "platformCode", "pageTypeCode", "timestamp", "pageParam", "holeCode", "accessCode")

    //获取订单详情
    val resultDataFrame = pageForward.filter(col("pageTypeCode").equalTo(PENDING_PAYMENT.code) || col("pageTypeCode").equalTo(COMPLETED_PAYMENT.code))
      .filter(!(col("holeCode").equalTo("kwnull") && col("accessCode").equalTo("null")))
      .select(commons.map(col): _*)
      .withColumn("categoryId", col("pageParam").getItem("categoryId"))
      .withColumn("productId", col("pageParam").getItem("productId"))
      .withColumn("orderNum", col("pageParam").getItem("orderNum"))
      .withColumn("paymentAmount", col("pageParam").getItem("paymentAmount"))
      .coalesce(20)

    val yesterday = Utils.getYesterday()

    //待支付
    var order = resultDataFrame.filter(col("pageTypeCode").equalTo(PENDING_PAYMENT.code))
      .orderBy(col("timestamp").desc)
      .select(col("platformCode"), col("channelCode"), col("holeCode"), col("accessCode"),
        when(col("platformCode").equalTo("2g53"), col("sessionId")).otherwise(col("deviceToken")).as("sessionId"),
        col("categoryId"), col("productId"),
        col("orderNum"), col("paymentAmount").as("order_amount"))
      .dropDuplicates("orderNum")

    //支付
    var pay = resultDataFrame.filter(col("pageTypeCode").equalTo(COMPLETED_PAYMENT.code))
      .orderBy(col("timestamp").desc)
      .select(col("orderNum").as("payNum"), col("paymentAmount").as("pay_amount"))
      .dropDuplicates("payNum")

    order.join(pay, order("orderNum") === pay("payNum"), "left_outer")
      .select(col("platformCode").as("platform_code"), col("channelCode").as("channel_code"),
        when(col("holeCode").equalTo("kwnull"), "").otherwise(col("holeCode")).as("hole_code"),
        when(col("accessCode").contains(TOPIC.code), col("accessCode")).otherwise("").as("access_code"),
        col("sessionId").as("session_id"), col("categoryId").as("category_id"),
        col("productId").as("product_id"), col("orderNum").as("order_num"), col("order_amount"), col("pay_amount"),
        lit(yesterday).as("oper_date"))
      .insertDF2MysqlDirect("flow_stat_pit_order")

  }
}
